package com.atguigu.gmall.realtime.common.function;

/*import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

public class HbaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection connection;
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(connection);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        // 提取待写入数据
        JSONObject dataJsonObj = value.f0;
        // 配置对象
        TableProcessDim tableProcessDim = value.f1;

        // 提取并移除type
        String type = dataJsonObj.remove("type").toString();
        // 提取SinkTable
        String sinkTable = tableProcessDim.getSinkTable();
        // 提取SinkColumns
        String sinkColumns = tableProcessDim.getSinkColumns();
        // 提取SinkFamily
        String sinkFamily = tableProcessDim.getSinkFamily();
        // 提取SinkRowKey
        // String sinkRowKey = tableProcessDim.getSinkRowKey(); 错误,不能这样提取,要的不是名字,要的是那个值
        // 先提取RowKey列,再提取对应的值
        String rowKeyColumn = tableProcessDim.getSinkRowKey();
        String rowKey = dataJsonObj.getString(rowKeyColumn);

        if (type.equals(Constant.MAXWELL_TYPE_DELETE)){
            // type : delete
            // 执行删除维度数据的操作
            HBaseUtil.deleteRow(connection, Constant.HBASE_NAMESPACE, sinkTable, rowKey);

        }else {
            // type : insert , update, bootstrap_insert
            // 执行写入维度数据的操作
            HBaseUtil.putRow(connection, Constant.HBASE_NAMESPACE, sinkTable, rowKey, sinkFamily, dataJsonObj);
        }

    }
}*/
/*import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

*//**
 * @author WEIYUNHUI
 * @date 2024/8/2 14:35
 *//*
public class HbaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection connection ;
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection() ;
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(connection);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        //提取待写入的数据
        JSONObject dataJsonObj = value.f0;
        //配置对象
        TableProcessDim tableProcessDim = value.f1;
        //提取并移除type
        String type = dataJsonObj.remove("type").toString() ;
        //提取sinkTable
        String sinkTable = tableProcessDim.getSinkTable();
        //提取列族
        String sinkFamily = tableProcessDim.getSinkFamily();

        //提取rowkey
        String rowkeyColumn = tableProcessDim.getSinkRowKey();
        String rowkey = dataJsonObj.getString(rowkeyColumn);

        if(Constant.MAXWELL_TYPE_DELETE.equals(type)){
            // type: delete
            // 执行删除维度数据的操作
            HBaseUtil.deleteRow(connection , Constant.HBASE_NAMESPACE , sinkTable , rowkey);
        }else{
            // type: bootstrap-insert  insert  update
            // 执行写入维度数据的操作
            HBaseUtil.putRow(connection , Constant.HBASE_NAMESPACE , sinkTable , rowkey , sinkFamily ,  dataJsonObj );
        }
    }
}*/
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author WEIYUNHUI
 * @date 2024/8/2 14:35
 */
public class HBaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {
    private Connection connection ;
    private Jedis jedis;
    @Override
    public void open(Configuration parameters) throws Exception {
        connection = HBaseUtil.getConnection() ;
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeConnection(connection);
            RedisUtil.closeJedis(jedis);
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> value, Context context) throws Exception {
        //提取待写入的数据
        JSONObject dataJsonObj = value.f0;
        //配置对象
        TableProcessDim tableProcessDim = value.f1;
        //提取并移除type
        String type = dataJsonObj.remove("type").toString() ;
        //提取sinkTable
        String sinkTable = tableProcessDim.getSinkTable();
        //提取列族
        String sinkFamily = tableProcessDim.getSinkFamily();

        //提取rowkey
        String rowKeyColumn = tableProcessDim.getSinkRowKey();
        String rowKey = dataJsonObj.getString(rowKeyColumn);

        if(Constant.MAXWELL_TYPE_DELETE.equals(type)){
            // type: delete
            // 执行删除维度数据的操作
            HBaseUtil.deleteRow(connection , Constant.HBASE_NAMESPACE , sinkTable , rowKey);
        }else{
            // type: bootstrap-insert  insert  update
            // 执行写入维度数据的操作
            HBaseUtil.putRow(connection , Constant.HBASE_NAMESPACE , sinkTable , rowKey , sinkFamily ,  dataJsonObj );
        }

        if (type.equals(Constant.MAXWELL_TYPE_DELETE) || type.equals(Constant.MAXWELL_TYPE_UPDATE)){
            RedisUtil.delDim(jedis, sinkTable, rowKey);
        }
    }
}